import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;

public class ProduceDataToKafka {
    public static void main(String[] args) throws IOException, InterruptedException {
        Map<String, Object> conf = getProducerConfiguration();
        KafkaProducer<String, String> producer = new KafkaProducer<>(conf);

        BufferedReader br = new BufferedReader(new FileReader("E:\\lagou_orders.json"));
        String json = br.readLine();
        while (json != null) {
            JSONObject jsonObject = JSON.parseObject(json);
            JSONArray products = jsonObject.getJSONArray("products");
            ProducerRecord<String, String> record = new ProducerRecord<>("stage7_module2_data", jsonObject.toJSONString());
            Future<RecordMetadata> result = producer.send(record);
            System.out.println(result);
            json = br.readLine();
            Thread.sleep(1000);
        }
    }

    private static Map<String, Object> getProducerConfiguration(){
        Map<String, Object> configs = new HashMap<>();
        configs.put("bootstrap.servers", "192.168.80.100:9092");
        configs.put("key.serializer", IntegerSerializer.class);
        configs.put("value.serializer", StringSerializer.class);
        return configs;
    }
}
